package com.atguigu.edu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.app.func.DimAsyncFunction;
import com.atguigu.edu.bean.InteractionUserCommentBean;
import com.atguigu.edu.util.ClickHouseUtil;
import com.atguigu.edu.util.DateFormatUtil;
import com.atguigu.edu.util.KafkaUtil;
import com.atguigu.edu.util.TimestampLtz3CompareUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;



public class DwsInteractionUserCommentWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop100:8020/edu");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
         */

        // TODO 3 读取kafka对应主题的数据dwd_interaction_comment
        String topicName = "dwd_interaction_comment";
        String groupID = "dws_interaction_user_comment_window";
        DataStreamSource<String> commentStream = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupID));



        // TODO 4. 转换数据结构
        SingleOutputStreamOperator<JSONObject> mappedStream = commentStream.map(JSON::parseObject);
        SingleOutputStreamOperator<InteractionUserCommentBean> javaBeanStream = mappedStream.map(
                jsonObj -> {
                    String userId = jsonObj.getString("user_id");
                    Long ts = jsonObj.getLong("ts") * 1000L;
                    InteractionUserCommentBean interactionUserCommentBean = InteractionUserCommentBean.builder()
                            .userIdSet(new HashSet<String>(
                                    Collections.singleton(userId)
                            ))
                            .userVv(1L)
                            .scoreSum(1L)
                            .ts(ts)
                            .build();
                    return interactionUserCommentBean;
                }
        );

        // TODO 6. 设置水位线
        SingleOutputStreamOperator<InteractionUserCommentBean> withWatermarkStream = javaBeanStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<InteractionUserCommentBean>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<InteractionUserCommentBean>() {
                                    @Override
                                    public long extractTimestamp(InteractionUserCommentBean javaBean, long recordTimestamp) {
                                        return javaBean.getTs();
                                    }
                                }
                        )
        );


        // TODO 7 按照课程分组

        KeyedStream<InteractionUserCommentBean, String> keyedByProIdStream =
                withWatermarkStream.keyBy(InteractionUserCommentBean::getCourseId);


        // TODO 8. 开窗
        WindowedStream<InteractionUserCommentBean, String, TimeWindow> windowDS = keyedByProIdStream.window(TumblingEventTimeWindows.of(
                org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)
        ));

        // TODO 9. 聚合计算
        SingleOutputStreamOperator<InteractionUserCommentBean> reducedStream = windowDS.reduce(
                new ReduceFunction<InteractionUserCommentBean>() {
                    @Override
                    public InteractionUserCommentBean reduce(InteractionUserCommentBean value1, InteractionUserCommentBean value2) throws Exception {
                        value1.setUserVv(value1.getUserVv() + value2.getUserVv());
                        value1.setScoreSum(value1.getScoreSum() + value2.getScoreSum());
                        return value1;
                    }
                },
                new ProcessWindowFunction<InteractionUserCommentBean, InteractionUserCommentBean, String, TimeWindow>() {
                    @Override
                    public void process(String key, Context context, Iterable<InteractionUserCommentBean> elements, Collector<InteractionUserCommentBean> out) throws Exception {
                        for (InteractionUserCommentBean element : elements) {
                            String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                            String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                            element.setStt(stt);
                            element.setEdt(edt);
                            element.setTs(System.currentTimeMillis());
                            out.collect(element);
                        }
                    }
                }
        );



        // TODO 10. 关联课程信息
        SingleOutputStreamOperator<InteractionUserCommentBean> fullInfoStream = AsyncDataStream.unorderedWait(
                reducedStream,
                new DimAsyncFunction<InteractionUserCommentBean>("dim_course_info".toUpperCase()) {

                    @Override
                    public void join(InteractionUserCommentBean javaBean, JSONObject jsonObj)  {
                        javaBean.setCourseName(jsonObj.getString("name"));
                    }

                    @Override
                    public String getKey(InteractionUserCommentBean javaBean) {
                        return javaBean.getCourseId();
                    }
                },
                100, TimeUnit.SECONDS
        );




        // TODO 11. 写入到 OLAP 数据库
        fullInfoStream.addSink(ClickHouseUtil.<InteractionUserCommentBean>getClickHouseSinkFunc(
                "insert into dws_interaction_user_comment_window values(?,?,?,?,?,?,?,?)"
        ));

        // TODO 12 执行任务
        env.execute(groupID);
    }
}
